#ifndef WIN32
#include <pthread.h>
#endif
#include <signal.h>
#include <staff/common/Exception.h>
#include <staff/common/Value.h>
#include <staff/common/DataObject.h>
#include <staff/common/DataObjectHelper.h>
#include <rise/common/StreamBuffer.h>
#include <rise/socket/RecvSocket.h>
#include <rise/threading/Thread.h>
#include "Message.h"
#include "MessageQueue.h"
#include "MessageQueueManager.h"
#include "ClientWorker.h"

namespace smq
{
  struct ClientWorker::ClientWorkerImpl: public rise::threading::CThread, public QueueWatcher
  {
    ClientWorkerImpl():
      bFinished(false)
    {
    }

    void Run(void* pParam)
    {
#ifdef OS_MCBC
      {
        sigset_t tSigMask;

        sigemptyset(&tSigMask);
        sigaddset(&tSigMask, SIGINT);
        pthread_sigmask (SIG_BLOCK, &tSigMask, NULL);
      }
#endif

      rise::LogEntry();
      bFinished = false;

      MessageQueueManager& rMessageQueueManager = MessageQueueManager::Inst();

      tSendBuffer.Reserve(4096);
      tRecvBuffer.Reserve(40960);

      while(!IsStopping())
      {
        staff::CDataObject tdoMsgIn;
        staff::CDataObject tdoMsgOut("");
        tdoMsgOut.SetNamespaceUri("http://tempui.org/staff-mq/messages/response");

        try
        {
          if (!Receive(tdoMsgIn))
          {
            break; // connection closed or thread stopping
          }

          const std::string& sMsgProc = tdoMsgIn.GetLocalName();

          tdoMsgOut.SetLocalName(sMsgProc + "Result");

          if (sMsgProc == "Ident")
          {
            sName = tdoMsgIn["sName"].AsString();
            sType = tdoMsgIn["sType"].AsString();
          }
          else
          // publisher msg
          if (sMsgProc == "Publish")
          {
            Message tMsg;
            tMsg.sId = tdoMsgIn["sId"].AsString();
            tMsg.sQueue = tdoMsgIn["sQueueName"].AsString();
            tMsg.sPublisher = tdoMsgIn["sPublisher"].AsString();
            tMsg.sText = tdoMsgIn.GetChildByLocalName("Message").Begin()->ToString();
            rMessageQueueManager.AddMessage(tMsg);
          }
          else
          if (sMsgProc == "CreateQueue")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            rMessageQueueManager.CreateQueue(sQueueName);
          }
          else
          if (sMsgProc == "RemoveQueue")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            rMessageQueueManager.RemoveQueue(sQueueName);
          }
          else
          // subscriber msg
          if (sMsgProc == "Subscribe")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            rMessageQueueManager.AddQueueWatcher(sQueueName, this);
          }
          else
          if (sMsgProc == "Unsubscribe")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            rMessageQueueManager.RemoveQueueWatcher(sQueueName, this);
          }
          else
          if (sMsgProc == "GetQueueList")
          {
            StringList lsQueues;
            rMessageQueueManager.GetQueueList(lsQueues);
            for (StringList::const_iterator itQueue = lsQueues.begin();
               itQueue != lsQueues.end(); ++itQueue)
            {
              staff::CDataObject tdoQueue = tdoMsgOut.CreateChild("Queue");
              tdoQueue.SetText(*itQueue);
            }
          }
          else
          if (sMsgProc == "GetFilteredQueueList")
          {
            const std::string sFilter = tdoMsgIn["sFilter"].AsString();
            StringList lsQueues;
            rMessageQueueManager.GetFilteredQueueList(sFilter, lsQueues);
            for (StringList::const_iterator itQueue = lsQueues.begin();
               itQueue != lsQueues.end(); ++itQueue)
            {
              staff::CDataObject tdoQueue = tdoMsgOut.CreateChild("Queue");
              tdoQueue.SetText(*itQueue);
            }
          }
          else
          if (sMsgProc == "GetQueueLength")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            int nLength = rMessageQueueManager.GetQueue(sQueueName).GetQueueLength();
            tdoMsgOut.SetValue(nLength);
          }
          else
          if (sMsgProc == "GetLastMessage")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            const MessageQueue& rMessageQueue = rMessageQueueManager.GetQueue(sQueueName);
            staff::CDataObject tdoMsg = tdoMsgOut.CreateChild("Message");
            tdoMsg << rMessageQueue.GetLastMessage();
          }
          else
          if (sMsgProc == "GetMessageHeadersList")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            const MessageQueue& rMessageQueue = rMessageQueueManager.GetQueue(sQueueName);
            MessageHeaderList lsHeaders;
            rMessageQueue.GetMessageHeadersList(lsHeaders);
            for (MessageHeaderList::const_iterator itHeader = lsHeaders.begin();
               itHeader != lsHeaders.end(); ++itHeader)
            {
              staff::CDataObject tdoHeader = tdoMsgOut.CreateChild("Header");
              tdoHeader << (*itHeader);
            }
          }
          else
          if (sMsgProc == "GetMessage")
          {
            const std::string sQueueName = tdoMsgIn["sQueueName"].AsString();
            const std::string sId = tdoMsgIn["sId"].AsString();
            const MessageQueue& rMessageQueue = rMessageQueueManager.GetQueue(sQueueName);
            const Message& rMsg = rMessageQueue.GetMessage(sId);
            staff::CDataObject tdoMessage = tdoMsgOut.CreateChild("Message");
            tdoMessage << rMsg;
          }
          else
          {
            rise::LogError() << "Unknown message proc: " << sMsgProc
                << "\n-----------------------------------\n" << tdoMsgIn.ToString()
                << "\n-----------------------------------\n";
            tdoMsgOut.SetLocalName("Fault");
            tdoMsgOut.SetNamespaceUri("http://tempui.org/staff-mq/messages/fault");
            tdoMsgOut.SetText("Unknown message proc: " + sMsgProc);
          }
        }
        catch(staff::CDomException& tException)
        {
          const std::string& sDescr = tException.GetString();
          rise::LogError() << "Exception: " << sDescr << " while parsing request:";

          tdoMsgOut.SetLocalName("Fault");
          tdoMsgOut.SetNamespaceUri("http://tempui.org/staff-mq/messages/fault");
          tdoMsgOut.SetText("Error while parsing request: " + sDescr);
        }
        catch(rise::CException& tException)
        {
          const std::string& sDescr = tException.GetString();
          rise::LogError() << "Exception: " << sDescr << "\nwhile processing message:"
                << "\n-----------------------------------\n" << tdoMsgIn.ToString()
                << "\n-----------------------------------\n";

          tdoMsgOut.SetLocalName("Fault");
          tdoMsgOut.SetNamespaceUri("http://tempui.org/staff-mq/messages/fault");
          tdoMsgOut.SetText("Error while executing proc: " + sDescr);
        }
        catch(...)
        {
          rise::LogError() << "Exception while processing message:"
                << "\n-----------------------------------\n" << tdoMsgIn.ToString()
                << "\n-----------------------------------\n";
          tdoMsgOut.SetLocalName("Fault");
          tdoMsgOut.SetNamespaceUri("http://tempui.org/staff-mq/messages/fault");
          tdoMsgOut.SetText("Unknown error while executing proc");
        }

        Post(tdoMsgOut);
      }

      if (sType == "subscriber")
      {
        rMessageQueueManager.RemoveQueueWatcher(this);
      }

      bFinished = true;
#ifndef WIN32
      pthread_kill(pthParent, SIGCHLD);
#endif
    }

    virtual void operator()(const Message& rMsg)
    {
      staff::CDataObject tdoMsg("Message");
      tdoMsg.SetNamespaceUri("http://tempui.org/staff-mq/messages/notify");
      tdoMsg << rMsg;
      Post(tdoMsg);
    }

    void Post(const staff::CDataObject& rMsg)
    {
      tSendBuffer.Reset();
      tSendBuffer << rMsg;

  #ifdef _DEBUG
      rise::LogDebug2() << "sending...";
  #endif

      // sending request
      rise::ulong ulSent = 0;
      rise::ulong ulBuffPos = 0;
      rise::ulong ulDataSize = tSendBuffer.GetSize();

#ifdef _DEBUG
        rise::LogDebug3() << "Sending: " << ulDataSize << " bytes:\n" <<
          "\033[34m" << std::string(reinterpret_cast<const char*>(tSendBuffer.GetData()), ulDataSize) << "\033[0m";
#endif
      while (ulDataSize > 0)
      {
        tSocket.Send(tSendBuffer.GetData() + ulBuffPos, ulDataSize, &ulSent);

        RISE_ASSERTES(ulSent, rise::CFileWriteException, "Unable to send");

        ulBuffPos += ulSent;
        ulDataSize -= ulSent;
      }
    }

    bool Receive(staff::CDataObject& rMsg)
    {
#ifdef _DEBUG
      rise::LogDebug2() << "receiving...";
#endif
      // receiving response
      bool bRcvd = true;
      rise::ulong ulCurrentSize = 0;
      rise::ulong ulReceived = tRecvBuffer.GetSize();
      tRecvBuffer.Normalize();
      while (bRcvd)
      {
        if (IsStopping())
        {
          return false;
        }

        if(memchr(tRecvBuffer.GetData() + ulCurrentSize, '\0', ulReceived) != NULL)
        {
          ulCurrentSize += ulReceived;
          break;
        }

        ulCurrentSize += ulReceived;

        bRcvd = tSocket.Recv(tRecvBuffer, true, &ulReceived);

        if (ulReceived == 0)
        {
          tSocket.Close();
          rise::LogDebug() << "Connection Closed";
          Stop();
          return false;
        }
      } // while (bRcvd)

#ifdef _DEBUG
      rise::LogDebug3() << "received: " << ulCurrentSize << " bytes:\n" <<
        "\033[35m" << std::string(reinterpret_cast<const char*>(tRecvBuffer.GetData()), ulCurrentSize) << "\033[0m";
#endif

      tRecvBuffer >> rMsg;

      return true;
    }

    rise::sockets::CRecvSocket tSocket;
    std::string sName;
    std::string sType;
#ifndef WIN32
    pthread_t pthParent;
#endif
    bool bFinished;

  private:
    rise::CStreamBuffer tSendBuffer;
    rise::CStreamBuffer tRecvBuffer;
  };


  ClientWorker::ClientWorker()
  {
    m_pImpl = new ClientWorkerImpl;
  }

  ClientWorker::~ClientWorker()
  {
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  rise::sockets::CRecvSocket& ClientWorker::GetSocket()
  {
    return m_pImpl->tSocket;
  }

  void ClientWorker::Start()
  {
#ifndef WIN32
    m_pImpl->pthParent = pthread_self();
#endif

    m_pImpl->Start();
  }

  void ClientWorker::Stop()
  {
    m_pImpl->Stop();
  }

  bool ClientWorker::IsFinished() const
  {
    return m_pImpl->bFinished;
  }

  void ClientWorker::Join()
  {
    int nRes = 0;
#ifndef WIN32
    pthread_t nId = m_pImpl->GetId();

    if (nId != 0)
    {
      while ((nRes = pthread_kill(nId, 0)) == 0)
      {
        rise::threading::CThread::Sleep(1);
      }
    }
#else
    m_pImpl->JoinThread();
#endif
  }

  const std::string& ClientWorker::GetName() const
  {
    return m_pImpl->sName;
  }

  const std::string& ClientWorker::GetType() const
  {
    return m_pImpl->sType;
  }

}
